/*
 * MIT License
 *
 * Copyright (c) 2020 wen.gu <454727014@qq.com>
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

 /***************************************************************************
 * Name: message.cpp
 *
 * Purpose: message and message queue imeplementation for 'COM' Functional
 *
 * Developer:
 *   wen.gu , 2020-07-29
 *
 * TODO:
 *
 ***************************************************************************/

 /******************************************************************************
 **    INCLUDES
 ******************************************************************************/
#define LOG_TAG "ICRQ"

#include "panda/com/runtime/ipc_client_requester.h"
#include "panda/com/runtime/ipc_types.h"
#include <future>

#include "panda/core/log.h"

namespace panda
{
namespace com
{
namespace runtime
{
/******************************************************************************
 **    MACROS
 ******************************************************************************/


/******************************************************************************
 **    VARIABLE DEFINITIONS
 ******************************************************************************/
using ClientResponseHandler = std::function<void(MessagePtr)>;

struct ClientReplyInfo
{
    std::string methodName = "";
    ClientResponseHandler handler = nullptr;
};

class IpcClientRequester::impl
{
public:
    using FunctionExecutor = panda::core::FunctionExecutor;
    using ClientAutoLock   = std::unique_lock<std::mutex>;
    using GuardAutoLock    = std::lock_guard<std::mutex>;
    using ClientSignalMap  = std::map<std::string, IpcClientSignalHandler>;
    using ClientReplyMap   = std::map<uint32_t, ClientReplyInfo>;
public:
    bool mIsTerminate = false;
    bool mIsConnected = false;
    uint32_t mSigTaskCnt = 0;
    uint32_t mRpyTaskCnt = 0;
    IpcClient* mClient = nullptr;
    IpcDataParser* mParser = nullptr;
    uint32_t mNextPacketId = 0;
    FunctionExecutor mSignalExecutor;
    FunctionExecutor mReplyExecutor;
    MessageQueue mReplyQueue; /** for receive repsonse information */
    MessageQueue mSignalQueue; /** for receive signal */
    std::mutex mApiLock;
    std::mutex mResponseLock;
    std::mutex mGeneratePacketIdLock;
    IpcPacketInfo mPktInfo;
    ClientSignalMap mSignals;
    ClientReplyMap mReplies;
public:
    impl();
    ~impl();
public:
    uint32_t generatePacketId();
    /** try to connect to service,
    *  if failed, then this function will be return immediately
    */
    PandaErrc connect(const std::string& srvAddr);

    /** disconnect from service */
    void disconnect();

    PandaErrc request(const std::string& method, const uint8_t* param, uint32_t size, IpcPacketData& reply, int32_t waitMs);

    PandaErrc requestAsync(const std::string& method, const uint8_t* param, uint32_t size, IpcClientReplyHandler rpyHandler);

    PandaErrc subscribeSignal(const std::string& sig, IpcClientSignalHandler sigHandler);

    PandaErrc unsubscribeSignal(const std::string& sig);

protected:
    bool onReply();
    bool onSignal();
    PandaErrc sendPacket(const std::string& name, uint32_t sessionId, MessageType type, const uint8_t* param, uint32_t size);

    /** return packet ID */
    uint32_t addReplyInfo(const std::string& name, ClientResponseHandler handler);

    void resubscribeSignal();

};
/******************************************************************************
 **    FUNCTION DEFINITIONS
 ******************************************************************************/

IpcClientRequester::impl::impl()
{
    /** in order to Ensure that signals are processed in the order they
        *  are sent, so only 1 thread to process signals
        */
    mSigTaskCnt = 1;
    mRpyTaskCnt = 1;
    mSignalExecutor.setFunction(std::bind(&impl::onSignal, this), mSigTaskCnt);
    mReplyExecutor.setFunction(std::bind(&impl::onReply, this), mRpyTaskCnt);

    mClient = new IpcClient([this](bool connected)
    {
        mIsConnected = connected;
        if (connected)
        {
            resubscribeSignal();
        }
    });

    mParser = new IpcDataParser([this](MessagePtr msg)
    {
        if (msg)
        {
            if (msg->type == MessageType::kResponse)
            {
                mReplyQueue.push(msg);
            }
            else if (msg->type == MessageType::kEvent)
            {
                mSignalQueue.push(msg);
            }
        }
    });
}

IpcClientRequester::impl::~impl()
{

}

uint32_t IpcClientRequester::impl::generatePacketId()
{
    GuardAutoLock al(mGeneratePacketIdLock);
    uint32_t ret = ++mNextPacketId;

    if (ret == 0)
    {
        ret = ++mNextPacketId;
    }

    return ret;
}

bool IpcClientRequester::impl::onReply()
{
    MessagePtr msg = mReplyQueue.popAndWait();

    if (msg)
    {
        GuardAutoLock gal(mResponseLock);
        ClientReplyMap::iterator it = mReplies.find(msg->sessionId);

        if (it != mReplies.end())
        {
            ClientReplyInfo& cpi = it->second;

            if ((cpi.methodName == msg->name) && cpi.handler)
            {
                cpi.handler(msg);
            }
        }
    }

    return !mIsTerminate;
}

bool IpcClientRequester::impl::onSignal()
{
    MessagePtr msg = mSignalQueue.popAndWait();

    if (msg)
    {
        ClientSignalMap::iterator it = mSignals.find(msg->name);

        if (it != mSignals.end())
        {
            IpcClientSignalHandler handler = it->second;

            if (handler)
            {
                IpcPacketData pktData;
                pktData.payload = msg->payload;
                pktData.size = msg->payloadSize;
                handler(msg->name, pktData);
            }
        }
    }

    return !mIsTerminate;
}

IpcClientRequester::PandaErrc IpcClientRequester::impl::sendPacket(const std::string& name, uint32_t sessionId, MessageType type, const uint8_t* param, uint32_t size)
{
    IpcPacketParam ipp;
    ipp.name = name;
    ipp.sessionId = sessionId;
    ipp.type = type;
    IpcPacketData pktData = mParser->makePacket(ipp, param, size);

    return mClient->sendData(pktData.payload.get(), pktData.size);
}

uint32_t IpcClientRequester::impl::addReplyInfo(const std::string& name, ClientResponseHandler handler)
{   
    uint32_t packetId = generatePacketId();
    GuardAutoLock gal(mResponseLock);    
    ClientReplyInfo& cpi  = mReplies[packetId];
    cpi.methodName = name;
    cpi.handler = handler;

    return packetId;
}

void IpcClientRequester::impl::resubscribeSignal()
{
    GuardAutoLock gal(mApiLock);
    uint32_t packetId = 0;
    for (auto it : mSignals)
    {
         packetId = generatePacketId();   

        PandaErrc ret = sendPacket(it.first, packetId, MessageType::kEventSubscribe, nullptr, 0);

        if (PandaErrc::OK == ret)
        {
            //todo

        }
    }
}

IpcClientRequester::PandaErrc IpcClientRequester::impl::connect(const std::string& srvAddr)
{
    return mClient->connect(srvAddr, [this](uint8_t* data, uint32_t size)
    {
        this->mParser->doParse(this->mPktInfo, data, size); /** todo something */
    }, 0);
}

/** disconnect from service */
void IpcClientRequester::impl::disconnect()
{
    mClient->disconnect();
    mIsConnected = false;
}

IpcClientRequester::PandaErrc IpcClientRequester::impl::request(const std::string& method, const uint8_t* param, uint32_t size, IpcPacketData& reply, int32_t waitMs)
{
    PandaErrc ret = PandaErrc::NoInit;
    if (mIsConnected)
    {
        std::shared_ptr<std::promise<IpcPacketData*> > promise = std::make_shared<std::promise<IpcPacketData*> >();
        std::future<IpcPacketData*> fut = promise->get_future();

        uint32_t packetId = addReplyInfo(method, [promise](MessagePtr msg)
        {
            IpcPacketData* crd = nullptr;
            if (msg && msg->payload)
            {
                crd = new IpcPacketData;
                crd->payload = msg->payload;
                crd->size = msg->payloadSize;
                //memcpy(crd->data, msg->payload(), crd->size);
            }

            promise->set_value(crd);
        });

        ret = sendPacket(method, packetId, MessageType::kRequest, param, size);


        if (ret == PandaErrc::OK)
        {
            IpcPacketData* rpy = nullptr;
            if (waitMs == COM_REQ_WAIT_INFINITE)
            {
                rpy = fut.get();
            }
            else
            {
                std::chrono::milliseconds span(waitMs);

                if (fut.wait_for(span) == std::future_status::ready)
                {
                    rpy = fut.get();
                }
                else
                {
                    return PandaErrc::Timeout;
                }
            }

            if (rpy)
            {
                reply = *rpy;
                delete rpy;
            }
            else
            {
                ret = PandaErrc::OperationCanceled;
            }
        }
        else
        {
            LOGE("send request(%s), packetId:%d, failed\n", method.c_str(), packetId);
        }
    }

    return ret;
}

IpcClientRequester::PandaErrc IpcClientRequester::impl::requestAsync(const std::string& method, const uint8_t* param, uint32_t size, IpcClientReplyHandler rpyHandler)
{
    PandaErrc ret = PandaErrc::NoInit;

    if (mIsConnected)
    {
        uint32_t packetId = addReplyInfo(method, [rpyHandler](MessagePtr msg)
        {
            if (msg)
            {
                if (rpyHandler)
                {
                    IpcPacketData pktData;
                    pktData.payload = msg->payload;
                    pktData.size = msg->payloadSize;
                    rpyHandler(msg->name, pktData);
                }
            }
            else
            {
                LOGW("reply with null Message\n");
            }
        });

        ret = sendPacket(method, packetId, MessageType::kRequest, param, size);

        if (PandaErrc::OK != ret)
        {
            LOGE("send request to binding failed(%s)\n", ErrorStr(ret));
        }
    }
    else
    {
        LOGE("client not connect to service \n");
    }

    return ret;
}

IpcClientRequester::PandaErrc IpcClientRequester::impl::subscribeSignal(const std::string& sig, IpcClientSignalHandler sigHandler)
{
    PandaErrc ret = PandaErrc::AlreadyExisted;
    GuardAutoLock gal(mApiLock);
    ClientSignalMap::iterator it = mSignals.find(sig);

    if (it == mSignals.end())
    {
        mSignals[sig] = sigHandler;
        ret = PandaErrc::OK;
    }

    return ret;
}

IpcClientRequester::PandaErrc IpcClientRequester::impl::unsubscribeSignal(const std::string& sig)
{
    PandaErrc ret = PandaErrc::AlreadyExisted;
    GuardAutoLock gal(mApiLock);

    ClientSignalMap::iterator it = mSignals.find(sig);

    if (it != mSignals.end())
    {
        uint32_t packetId = 0;
        {
            GuardAutoLock gal(mResponseLock);
            packetId = generatePacketId();
        }

        ret = sendPacket(it->first, packetId, MessageType::kEventUnsubscribe, nullptr, 0);

        if (PandaErrc::OK == ret)
        {
            mSignals.erase(it);
        }
    }

    return ret;
}
////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////



IpcClientRequester::IpcClientRequester()
    :mImpl(new impl)
{


   /** todo something */
}

IpcClientRequester::~IpcClientRequester()
{

    /** todo something */
}


/** try to connect to service,
*  if failed, then this function will be return immediately
*/
IpcClientRequester::PandaErrc IpcClientRequester::connect(const std::string& srvAddr)
{
    return mImpl->connect(srvAddr);

}

/** disconnect from service */
void IpcClientRequester::disconnect()
{
    return mImpl->disconnect();
}

IpcClientRequester::PandaErrc IpcClientRequester::request(const std::string& method, const uint8_t* param, uint32_t size, IpcPacketData& reply, int32_t waitMs /*= COM_REQ_WAIT_INFINITE*/)
{
    return mImpl->request(method, param, size, reply, waitMs);
}

IpcClientRequester::PandaErrc IpcClientRequester::requestAsync(const std::string& method, const uint8_t* param, uint32_t size, IpcClientReplyHandler rpyHandler)
{
    return mImpl->requestAsync(method, param, size, rpyHandler);
}


IpcClientRequester::PandaErrc IpcClientRequester::subscribeSignal(const std::string& sig, IpcClientSignalHandler sigHandler)
{
    return mImpl->subscribeSignal(sig, sigHandler);
}

IpcClientRequester::PandaErrc IpcClientRequester::unsubscribeSignal(const std::string& sig)
{
    return mImpl->unsubscribeSignal(sig);
}


} /** namespace runtime */
} /** namespace com */
} /** namespace panda */


